package com.itgacl.mqtt.client;

import com.itgacl.mqtt.client.config.MqttClientConfig;
import com.itgacl.mqtt.client.util.IpUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.util.StringUtils;

import java.nio.charset.StandardCharsets;
import java.util.Optional;
import java.util.UUID;

/**
 * mqttClientService
 */
@Slf4j
public class MqttClientService implements MqttCallbackExtended {

    private MqttAsyncClient mqttAsyncClient;
    private MqttConnectOptions clientOptions;
    private final Object connectLock = new Object();

    private final MqttClientConfig mqttClientConfig;

    private final IMqttMessageListener mqttMessageListener;

    public MqttClientService(MqttClientConfig mqttClientConfig,IMqttMessageListener mqttMessageListener){
        this.mqttClientConfig = mqttClientConfig;
        this.mqttMessageListener = mqttMessageListener;
        log.debug("初始化mqtt客户端");
        initMqttClient();
    }

    /**
     * 销毁服务
     *
     * @throws Exception
     */
    public void destroy() throws Exception {
        mqttAsyncClient.disconnect();
    }


    //初始化mqtt客户端
    private void initMqttClient() {
        try {
            String clientId;
            if(!StringUtils.isEmpty(mqttClientConfig.getClientId())){
                //连接的clientId必须唯一
                clientId = mqttClientConfig.getClientId() + "-" + IpUtil.getLocalIP();
            }else {
                //随机生成一个clientId
                clientId = UUID.randomUUID()+ "-" + IpUtil.getLocalIP();
            }
            log.debug("mqttClientId:{}", clientId);
            mqttAsyncClient = new MqttAsyncClient("tcp://" + mqttClientConfig.getConnHost() + ":" + mqttClientConfig.getConnPort(),
                    clientId, new MemoryPersistence());
            mqttAsyncClient.setCallback(this);

            //初始化 MQTT 连接服务器配置参数对象,以下的配置顺序将按照 MQTT 3.1.1 版本的各参数先后顺序进行配置 (bit 从 0 - 7)
            clientOptions = new MqttConnectOptions();

            //设置 MQTT 通信协议版本
//            clientOptions.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);

            //设置 MQTT协议 -[连接服务端] - [连接标志/Connect Flags] - [清理会话标志/Clean Session]（可变报头第 8 个字节 第 1 位）
            clientOptions.setCleanSession(true);

            //暂定第 2 位的 [遗嘱标志/Will Flag] 暂时不设置,因此 第 3 位，第 4 位的 [遗嘱服务质量等级/Will QoS]不设置,第 5 位的 [遗嘱保留/Will Retain]
            //这里是服务器端，等需发遗嘱消息的场景扩展

            //设置 MQTT协议 -[连接服务端] - [连接标志/Connect Flags] - [登陆名称标志/User Name Flag]（可变报头第 8 个字节 第 7 位）
            //设置 MQTT协议 -[连接服务端] - [连接标志/Connect Flags] - [登陆密码标志/Password Flag]（可变报头第 8 个字节 第 6 位）
            //因为 MQTT 客户端的连接服务端请求中的 “登陆名称” 和 “登陆密码” 可以不存在(只有设置才存在,不设置压根不存在,所以不存在空),因此需要判断有无提供名称和密码
            if (mqttClientConfig.getUsername() != null) {
                //给定了 “登陆名称” ,将其设置
                clientOptions.setUserName(mqttClientConfig.getUsername());
                //给定了 “登陆密码” ,将其设置 （MQTT 连接服务端规定允许有“登陆名称”,但却没有“登陆密码”,但是如果没有“登陆名称”,一定不存在“登陆密码”,没有为什么,MQTT 协议规定）
                if (mqttClientConfig.getPassword() != null) {
                    clientOptions.setPassword(mqttClientConfig.getPassword().toCharArray());
                }
            }

            //设置 MQTT协议 -[连接服务端] - [保持连接/Keep Alive] - 发送请求 间隔的周期 ,第 9(MSB) 第 10(LSB) 个字节,API 默认以“秒”为计时单位
            clientOptions.setKeepAliveInterval(60);
            //设置 MQTT协议 -[连接服务端] - [保持连接/Keep Alive] - 每次发送请求的超时时间, API 默认以“秒”为计时单位
            clientOptions.setConnectionTimeout(30);

            //最后设置是否支持与 MQTT 服务器端连接断开后自动重连
            clientOptions.setAutomaticReconnect(true);
            checkConnection();
        } catch (MqttException e) {
            log.error("[{}:{}] MQTT broker connection failed!", mqttClientConfig.getConnHost(), mqttClientConfig.getConnPort(), e);
            throw new RuntimeException("MQTT broker connection failed!", e);
        }
    }

    private void checkConnection() {
        if (!mqttAsyncClient.isConnected()) {
            synchronized (connectLock) {
                while (!mqttAsyncClient.isConnected()) {
                    log.debug("[{}:{}] MQTT broker connection attempt!", mqttClientConfig.getConnHost(), mqttClientConfig.getConnPort());
                    try {
                        mqttAsyncClient.connect(clientOptions, null, new IMqttActionListener() {
                            @Override
                            public void onSuccess(IMqttToken iMqttToken) {
                                log.info("[{}:{}] MQTT broker connection established!", mqttClientConfig.getConnHost(), mqttClientConfig.getConnPort());
                            }

                            @Override
                            public void onFailure(IMqttToken iMqttToken, Throwable e) {
                                log.warn("[{}:{}] MQTT broker connection faied! {}", mqttClientConfig.getConnHost(), mqttClientConfig.getConnPort(), e.getMessage(), e);
                            }
                        }).waitForCompletion();

                    } catch (MqttException e) {
                        log.warn("[{}:{}] MQTT broker connection failed!", mqttClientConfig.getConnHost(), mqttClientConfig.getConnPort(), e);
                        if (!mqttAsyncClient.isConnected()) {
                            try {
                                Thread.sleep(mqttClientConfig.getRetryInterval());
                            } catch (InterruptedException e1) {
                                log.trace("Failed to wait for retry interval!", e);
                            }
                        }
                    }
                }
            }
        }
    }

    /**
     * 订阅主题
     *
     * @param topic
     * @param mqttMessageListener
     * @throws MqttException
     */
    public void subscribe(String topic, IMqttMessageListener mqttMessageListener) throws MqttException {
        IMqttToken token = mqttAsyncClient.subscribe(topic, 1, mqttMessageListener);
        token.waitForCompletion();
    }

    public void subscribe(IMqttMessageListener mqttMessageListener) throws MqttException {
        Optional.of(mqttClientConfig.getSubscribeTopic())
                .ifPresent(topicList -> topicList.forEach(topicEle -> {
                    try {
                        log.info("MqttClientServiceImpl subscribe订阅主题：{}", topicEle);
                        IMqttToken token = mqttAsyncClient.subscribe(topicEle, 1);
                        token.waitForCompletion();
                    } catch (MqttException e) {
                        log.error("批量订阅消息主题遭遇异常 -> {}", e.getMessage());
                    }
                }));
    }


    /**
     * 发布消息
     *
     * @param topic
     * @param msg
     */
    public void publish(String topic, String msg) {
        try {
            MqttMessage mqttMessage = new MqttMessage(msg.getBytes(StandardCharsets.UTF_8));
            mqttAsyncClient.publish(topic, mqttMessage, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken iMqttToken) {
                    log.info("Successfully published to topic [{}]", topic);
                }

                @Override
                public void onFailure(IMqttToken iMqttToken, Throwable e) {
                    log.warn("Failed to publish to topic [{}]", topic, e);
                }
            });
        } catch (MqttException e) {
            log.warn("Failed to publish to topic [{}] ", topic, e);
        }
    }


    /**
     * 发布消息
     *
     * @param topic
     * @param msg
     */
    public void publish(String topic, byte[] msg) {
        try {
            MqttMessage mqttMessage = new MqttMessage(msg);
            mqttAsyncClient.publish(topic, mqttMessage, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken iMqttToken) {
                    log.info("Successfully published to topic [{}]", topic);
                }

                @Override
                public void onFailure(IMqttToken iMqttToken, Throwable e) {
                    log.warn("Failed to publish to topic [{}]", topic, e);
                }
            });
        } catch (MqttException e) {
            log.warn("Failed to publish to topic [{}] ", topic, e);
        }
    }

    @Override
    public void connectionLost(Throwable throwable) {
        log.info("连接丢失了，丢失原因 -> {}", throwable.getMessage());
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        log.info("MqttClientService messageArrived，topic：{},MqttMessage:{}", topic, mqttMessage);
        mqttMessageListener.messageArrived(topic, mqttMessage);
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        //消息发送完成后的回调。
        //消息发送完成后的分布式逻辑，是通过这个回调触发的。
    }


    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        log.info("调用了connectComplete，是否重连 = {} ,serverURI = {} ", reconnect, serverURI);
        try {
            subscribe(mqttMessageListener);//订阅主题
        } catch (MqttException e) {
            log.error(e.getMessage(), e);
        }
    }


}
